﻿//
// © Algobox Ltd - Benn Eichhorn
//
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections.Concurrent;
using Algobox.Objects.Finance.Assets.Stock;
using Algobox.Feeds.Finance.MarketData;
using System.IO;
using System.IO.Compression;
using Algobox.Feeds.Finance.Bats.PitchClient;
using Algobox.Structure.Finance.Prices;
using Algobox.Objects.Finance.Prices;

namespace Algobox.Feeds.Finance.Bats.Adaptors
{
    public class PitchFileMarketFeedAdaptor : IMarketFeedAdaptor, IBatsMessageHandler
    {
        /// <summary>
        /// Process a pitch market data file
        /// Only partially thread safe
        /// </summary>
        public PitchFileMarketFeedAdaptor()
        {
            _batsPitchClient = new BatsPitchClient(this);
            _isProcessingFile = false;
        }

        /// <summary>
        /// Number of bytes to read from file
        /// </summary>
        const int BUFFERSIZE = 1024;

        /// <summary>
        /// Message termination character
        /// </summary>
        //const char TERMINATOR = '\x0A';
        const char TERMINATOR = '\n';

        const int MAX_QUEUE = 1000;

        readonly TimeSpan EOD = new TimeSpan(16, 29, 59);

        BatsPitchClient _batsPitchClient;

        ConcurrentDictionary<string, StockWatch> _mapDepthOrderRef = new ConcurrentDictionary<string, StockWatch>();
        ConcurrentDictionary<string, StockWatch> _mapDepthSymbol = new ConcurrentDictionary<string, StockWatch>();
        ConcurrentDictionary<IStock, int> _mapSecuritiesCount = new ConcurrentDictionary<IStock, int>();

        ConcurrentDictionary<IStock, List<IMarketUpdateAsk>> _mapAskRequests = new ConcurrentDictionary<IStock, List<IMarketUpdateAsk>>();
        ConcurrentDictionary<IStock, List<IMarketUpdateBid>> _mapBidRequests = new ConcurrentDictionary<IStock, List<IMarketUpdateBid>>();
        ConcurrentDictionary<IStock, List<IMarketUpdateLast>> _mapLastRequests = new ConcurrentDictionary<IStock, List<IMarketUpdateLast>>();
        ConcurrentDictionary<IStock, List<IMarketUpdatePhase>> _mapPhaseRequests = new ConcurrentDictionary<IStock, List<IMarketUpdatePhase>>();

        private bool _isProcessingFile;

        public event Action<IStock, long> OnSubscriptionAdded;
        public event Action<IStock, long> OnSubscriptionRemoved;
        public event Action OnMarketFeedShutdown;


        public string FeedName { get { return "Bats PITCH File Adaptor"; } }
        public bool IsStarted { get; set; }


        public bool Start()
        {
            IsStarted = true;
            return true;
        }

        public void Stop()
        {
            IsStarted = false;
            _batsPitchClient.Dispose();
        }

        /// <summary>
        /// Process a batch pitch file
        /// Not thread safe
        /// </summary>
        /// <param name="file"></param>
        public void ProcessFile(FileInfo file)
        {
            if (!_isProcessingFile)
            {
                _isProcessingFile = true;
                InitialiseStocks();

                byte[] bytes = new byte[BUFFERSIZE];

                ASCIIEncoding encoder = new ASCIIEncoding();

                string message = String.Empty;

                int readLength;
                int messageLength;

                // Get the stream of the source file.
                using (FileStream inFile = file.OpenRead())
                {
                    using (GZipStream decompress = new GZipStream(inFile, CompressionMode.Decompress))
                    {
                        while (IsStarted && decompress.CanRead && decompress.BaseStream.Position < decompress.BaseStream.Length)
                        {
                            if (_batsPitchClient.Count > MAX_QUEUE)
                            {
                                System.Threading.Thread.Sleep(100);
                            }
                            else
                            {
                                // read from the socket in the bytes buffer
                                readLength = decompress.Read(bytes, 0, BUFFERSIZE);

                                if (readLength > 0)
                                {
                                    // join the new buffer to the end of the last message
                                    message = String.Concat(message, encoder.GetString(bytes, 0, readLength));

                                    // find the end of the next message
                                    messageLength = message.IndexOf(TERMINATOR);

                                    while (messageLength > 0)
                                    {
                                        // enqueue the message
                                        _batsPitchClient.EnqueueMessage(message.Substring(0, messageLength));

                                        // remove message + the terminator charactor
                                        message = message.Remove(0, messageLength + 1);

                                        // find the end of the next message if it exists
                                        messageLength = message.IndexOf(TERMINATOR);
                                    }
                                }
                            }
                        }
                    }
                }
                SendClosePrices();
                _isProcessingFile = false;
            }
        }

        private void SendClosePrices()
        {
            PriceVolume last;
            List<IMarketUpdateLast> callbacks;
            foreach (IStock stock in _mapLastRequests.Keys.Where(a => a.Prices.Last.TimeStamp.TimeOfDay < EOD))
            {
                if (_mapLastRequests.TryGetValue(stock, out callbacks))
                {
                    last = stock.Prices.Last;
                    last.TimeStamp = last.TimeStamp.Date + EOD;
                    callbacks.ForEach(callback => callback.OnPriceLast(stock, last));
                }
            }
        }

        private void InitialiseStocks()
        {
            _mapDepthSymbol.Values.ToList().ForEach(
                (stockWatch) =>
                {
                    stockWatch.Bid.Price = 0;
                    stockWatch.Bid.Volume = 0;
                    stockWatch.Ask.Price = 0;
                    stockWatch.Ask.Volume = 0;
                    stockWatch.Last.Price = 0;
                    stockWatch.Last.Volume = 0;
                    stockWatch.OHLC.Open = 0;
                    stockWatch.OHLC.High = 0;
                    stockWatch.OHLC.Low = 0;
                    stockWatch.OHLC.Volume = 0;
                    stockWatch.VWAP = 0;
                });
            _mapSecuritiesCount.Keys.ToList().ForEach(
                (stock) =>
                {
                    stock.Phase = Objects.Finance.Phases.Phase.Open;
                });
        }

        public void ProcessMessage(PitchClient.Messages.IMessage message)
        {
            if (message is PitchClient.Messages.OrderAdd)
            {
                PitchClient.Messages.OrderAdd order = (PitchClient.Messages.OrderAdd)message;
                StockWatch stockWatch;
                if (_mapDepthSymbol.TryGetValue(order.Symbol, out stockWatch))
                {
                    PriceVolume best;
                    _mapDepthOrderRef.TryAdd(order.OrderReference, stockWatch);
                    if (stockWatch.Depth.OrderAdd(order.OrderReference, order.IsBuy, order.Price, order.Shares, order.TimeStamp, out best))
                    {
                        if (best.Type == PriceVolume.PriceType.Bid)
                        {
                            List<IMarketUpdateBid> list;
                            if (_mapBidRequests.TryGetValue(stockWatch.Stock, out list))
                            {
                                list.ForEach(callback => callback.OnPriceBid(stockWatch.Stock, best));
                            }
                        }
                        else
                        {
                            List<IMarketUpdateAsk> list;
                            if (_mapAskRequests.TryGetValue(stockWatch.Stock, out list))
                            {
                                list.ForEach(callback => callback.OnPriceAsk(stockWatch.Stock, best));
                            }
                        }
                    }
                }
            }
            else if (message is PitchClient.Messages.OrderExecution)
            {
                PitchClient.Messages.OrderExecution order = (PitchClient.Messages.OrderExecution)message;
                StockWatch stockWatch;
                if (_mapDepthOrderRef.TryGetValue(order.OrderReference, out stockWatch))
                {
                    bool isComplete;
                    PriceVolume best, last;
                    if (stockWatch.Depth.OrderExecution(order.OrderReference, order.ExecutedShares, order.TimeStamp, out last, out best, out isComplete))
                    {
                        if (best.Type == PriceVolume.PriceType.Bid)
                        {
                            List<IMarketUpdateBid> list;
                            if (_mapBidRequests.TryGetValue(stockWatch.Stock, out list))
                            {
                                list.ForEach(callback => callback.OnPriceBid(stockWatch.Stock, best));
                            }
                        }
                        else
                        {
                            List<IMarketUpdateAsk> list;
                            if (_mapAskRequests.TryGetValue(stockWatch.Stock, out list))
                            {
                                list.ForEach(callback => callback.OnPriceAsk(stockWatch.Stock, best));
                            }
                        }
                    }

                    if (last != null)
                    {
                        List<IMarketUpdateLast> callbacks;
                        if (_mapLastRequests.TryGetValue(stockWatch.Stock, out callbacks))
                        {
                            callbacks.ForEach(callback => callback.OnPriceLast(stockWatch.Stock, last));
                        }
                    }

                    if (isComplete)
                    {
                        _mapDepthOrderRef.TryRemove(order.OrderReference, out stockWatch);
                    }
                }
            }
            else if (message is PitchClient.Messages.OrderCancel)
            {
                PitchClient.Messages.OrderCancel order = (PitchClient.Messages.OrderCancel)message;
                StockWatch stockWatch;
                if (_mapDepthOrderRef.TryGetValue(order.OrderReference, out stockWatch))
                {
                    bool isComplete;
                    PriceVolume best;
                    if (stockWatch.Depth.OrderCancel(order.OrderReference, order.CanceledShares, order.TimeStamp, out best, out isComplete))
                    {
                        if (best.Type == PriceVolume.PriceType.Bid)
                        {
                            List<IMarketUpdateBid> list;
                            if (_mapBidRequests.TryGetValue(stockWatch.Stock, out list))
                            {
                                list.ForEach(callback => callback.OnPriceBid(stockWatch.Stock, best));
                            }
                        }
                        else
                        {
                            List<IMarketUpdateAsk> list;
                            if (_mapAskRequests.TryGetValue(stockWatch.Stock, out list))
                            {
                                list.ForEach(callback => callback.OnPriceAsk(stockWatch.Stock, best));
                            }
                        }
                    }
                    if (isComplete)
                    {
                        _mapDepthOrderRef.TryRemove(order.OrderReference, out stockWatch);
                    }
                }
            }
            else if (message is PitchClient.Messages.Trade)
            {
                PitchClient.Messages.Trade trade = (PitchClient.Messages.Trade)message;
                StockWatch stockWatch;
                if (_mapDepthSymbol.TryGetValue(trade.Symbol, out stockWatch))
                {
                    stockWatch.Last.Price = trade.Price;
                    stockWatch.Last.Volume = trade.Shares;
                    stockWatch.Last.TimeStamp = trade.TimeStamp;

                    List<IMarketUpdateLast> callbacks;
                    if (_mapLastRequests.TryGetValue(stockWatch.Stock, out callbacks))
                    {
                        callbacks.ForEach(callback => callback.OnPriceLast(stockWatch.Stock, stockWatch.Last));
                    }
                }
            }
        }


        public void MessageFailure(PitchClient.Messages.IMessage message)
        {

        }

        public void SubscribeBid(IStock stock, IMarketUpdateBid callback)
        {
            int requestCount = 0;
            _mapBidRequests.AddOrUpdate(stock,
                    (security) =>
                    {
                        requestCount = _mapSecuritiesCount.AddOrUpdate(security,
                            (key) =>
                            {
                                _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                                return 1;
                            },
                            (key, count) =>
                            {
                                return count + 1;
                            });
                        return new List<IMarketUpdateBid>() { callback };
                    },
                    (security, list) =>
                    {
                        if (!list.Contains(callback))
                        {
                            requestCount = _mapSecuritiesCount.AddOrUpdate(security,
                               (key) =>
                               {
                                   _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                                   return 1;
                               },
                               (key, count) =>
                               {
                                   return count + 1;
                               });
                            list.Add(callback);
                        }
                        else { } // don't subscribe twice you twat!
                        return list;
                    }
                );
            if (OnSubscriptionAdded != null)
            {
                OnSubscriptionAdded(stock, requestCount);
            }
        }

        public void SubscribeAsk(IStock stock, IMarketUpdateAsk callback)
        {
            int requestCount = 0;
            _mapAskRequests.AddOrUpdate(stock,
                (security) =>
                {
                    requestCount = _mapSecuritiesCount.AddOrUpdate(security,
                        (key) =>
                        {
                            _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                            return 1;
                        },
                        (key, count) =>
                        {
                            return count + 1;
                        });
                    return new List<IMarketUpdateAsk>() { callback };
                },
                (security, list) =>
                {
                    if (!list.Contains(callback))
                    {
                        requestCount = _mapSecuritiesCount.AddOrUpdate(security,
                            (key) =>
                            {
                                _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                                return 1;
                            },
                            (key, count) =>
                            {
                                return count + 1;
                            });
                        list.Add(callback);
                    }
                    else { } // don't subscribe twice you twat!
                    return list;
                }
            );
            if (OnSubscriptionAdded != null)
            {
                OnSubscriptionAdded(stock, requestCount);
            }
        }

        public void SubscribeLast(IStock stock, IMarketUpdateLast callback)
        {
            int requestCount = 0;
            _mapLastRequests.AddOrUpdate(stock,
                (security) =>
                {
                    requestCount = _mapSecuritiesCount.AddOrUpdate(security,
                        (key) =>
                        {
                            _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                            return 1;
                        },
                        (key, count) =>
                        {
                            return count + 1;
                        });
                    return new List<IMarketUpdateLast>() { callback };
                },
                (security, list) =>
                {
                    if (!list.Contains(callback))
                    {
                        requestCount = _mapSecuritiesCount.AddOrUpdate(security,
                            (key) =>
                            {
                                _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                                return 1;
                            },
                            (key, count) =>
                            {
                                return count + 1;
                            });
                        list.Add(callback);
                    }
                    else { } // don't subscribe twice you twat!
                    return list;
                }
            );
            if (OnSubscriptionAdded != null)
            {
                OnSubscriptionAdded(stock, requestCount);
            }
        }

        public void SubscribePhase(IStock stock, IMarketUpdatePhase callback)
        {
            int requestCount = 0;
            _mapPhaseRequests.AddOrUpdate(stock,
                (security) =>
                {
                    requestCount = _mapSecuritiesCount.AddOrUpdate(security,
                        (key) =>
                        {
                            _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                            return 1;
                        },
                        (key, count) =>
                        {
                            return count + 1;
                        });
                    return new List<IMarketUpdatePhase>() { callback };
                },
                (security, list) =>
                {
                    if (!list.Contains(callback))
                    {
                        requestCount = _mapSecuritiesCount.AddOrUpdate(security,
                            (key) =>
                            {
                                _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                                return 1;
                            },
                            (key, count) =>
                            {
                                return count + 1;
                            });
                        list.Add(callback);
                    }
                    else { } // don't subscribe twice you twat!
                    return list;
                }
            );
            if (OnSubscriptionAdded != null)
            {
                OnSubscriptionAdded(stock, requestCount);
            }
        }

        public void UnsubscribeBid(IStock stock, IMarketUpdateBid callback)
        {
            _mapBidRequests.AddOrUpdate(stock,
                (security) =>
                {
                    // shouldn't ever really come in here
                    _mapSecuritiesCount.AddOrUpdate(security,
                        (key) =>
                        {
                            _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                            return 0;
                        },
                        (key, count) =>
                        {
                            return count;
                        });
                    return new List<IMarketUpdateBid>();
                },
                (security, list) =>
                {
                    if (list.Contains(callback))
                    {
                        _mapSecuritiesCount.AddOrUpdate(security,
                            (key) =>
                            {
                                _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                                return 0;
                            },
                            (key, count) =>
                            {
                                --count;
                                if (OnSubscriptionRemoved != null)
                                {
                                    OnSubscriptionRemoved(key, count);
                                }
                                return count;
                            });
                        list.Remove(callback);
                    }
                    else { } // no subscription found, don't do that!  
                    return list;
                });
        }

        public void UnsubscribeAsk(IStock stock, IMarketUpdateAsk callback)
        {
            _mapAskRequests.AddOrUpdate(stock,
                (security) =>
                {
                    // shouldn't ever really come in here
                    _mapSecuritiesCount.AddOrUpdate(security,
                        (key) =>
                        {
                            _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                            return 0;
                        },
                        (symbol, count) =>
                        {
                            return count;
                        });
                    return new List<IMarketUpdateAsk>();
                },
                (security, list) =>
                {
                    if (list.Contains(callback))
                    {
                        _mapSecuritiesCount.AddOrUpdate(security,
                            (key) =>
                            {
                                _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                                return 0;
                            },
                            (key, count) =>
                            {
                                --count;
                                if (OnSubscriptionRemoved != null)
                                {
                                    OnSubscriptionRemoved(key, count);
                                }
                                return count;
                            });
                        list.Remove(callback);
                    }
                    else { } // no subscription found, don't do that!  
                    return list;
                });
        }

        public void UnsubscribeLast(IStock stock, IMarketUpdateLast callback)
        {
            if (stock is StockSecurity)
            {
                _mapLastRequests.AddOrUpdate(stock,
                    (security) =>
                    {
                        // shouldn't ever really come in here
                        _mapSecuritiesCount.AddOrUpdate(security,
                            (key) =>
                            {
                                _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                                return 0;
                            },
                            (key, count) =>
                            {
                                return count;
                            });
                        return new List<IMarketUpdateLast>();
                    },
                    (security, list) =>
                    {
                        if (list.Contains(callback))
                        {
                            _mapSecuritiesCount.AddOrUpdate(security,
                                (key) =>
                                {
                                    _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                                    return 0;
                                },
                                (key, count) =>
                                {
                                    --count;
                                    if (OnSubscriptionRemoved != null)
                                    {
                                        OnSubscriptionRemoved(key, count);
                                    }
                                    return count;
                                });
                            list.Remove(callback);
                        }
                        else { } // no subscription found, don't do that!  
                        return list;
                    });
            }
            else
            {
                // need to implement listing requests
            }
        }

        public void UnsubscribePhase(IStock stock, IMarketUpdatePhase callback)
        {
            _mapPhaseRequests.AddOrUpdate(stock,
                (security) =>
                {
                    // shouldn't ever really come in here
                    _mapSecuritiesCount.AddOrUpdate(security,
                        (key) =>
                        {
                            _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                            return 0;
                        },
                        (symbol, count) =>
                        {
                            return count;
                        });
                    return new List<IMarketUpdatePhase>();
                },
                (security, list) =>
                {
                    if (list.Contains(callback))
                    {
                        _mapSecuritiesCount.AddOrUpdate(security,
                            (key) =>
                            {
                                _mapDepthSymbol.TryAdd(key.Symbol, key.Prices);
                                return 0;
                            },
                            (key, count) =>
                            {
                                --count;
                                if (OnSubscriptionRemoved != null)
                                {
                                    OnSubscriptionRemoved(key, count);
                                }
                                return count;
                            });
                        list.Remove(callback);
                    }
                    else { } // no subscription found, don't do that!  
                    return list;
                });
        }



        public void SubscribeOpen(IStock stock, IMarketUpdateOpen callback)
        {
            throw new NotImplementedException();
        }

        public void UnsubscribeOpen(IStock stock, IMarketUpdateOpen callback)
        {
            throw new NotImplementedException();
        }
    }
}
